package com.zifeng.canal;

import com.alibaba.fastjson.JSON;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.zifeng.constant.RedisConstants;
import com.zifeng.pojo.User;
import com.zifeng.utils.CanalMessageToObject;
import com.zifeng.utils.RedisUtils;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import java.net.InetSocketAddress;
import java.util.List;


@Slf4j
//@Component
public class UserCanalEventHandler {

    private boolean executed = false;

    private final static int BATCH_SIZE = 200;

    private CanalConnector connector;


    @PostConstruct
    public void init() {
        connector = CanalConnectors.newSingleConnector(
                new InetSocketAddress("127.0.0.1", 11111),
                "user", "", "");

        try {
            connector.connect();
            connector.subscribe("user");
            connector.rollback();
        } catch (Exception e) {
            log.error("user canal 初始化异常: {}", e.getMessage());
            throw new RuntimeException(e);
        }
    }

    @PreDestroy
    public void onDestroy() {
        log.info("UserCanalEventHandler Bean 正在销毁，关闭 CanalConnector");
        if (connector != null) {
            try {
                connector.disconnect();  // 优雅断开连接
                log.info("CanalConnector 已断开连接");
            } catch (Exception e) {
                log.error("user canal 销毁异常: {}", e.getMessage());
                throw new RuntimeException(e);
            }
        }
    }

    //每2秒执行一次，扫描user数据表的数据变化
    @Scheduled(initialDelay = 2000, fixedDelay = Long.MAX_VALUE)
    public void handleCanalMessage() {
        if(!executed) {
            while (true) {
                // 获取指定数量的数据
                Message message = connector.getWithoutAck(BATCH_SIZE);
                //获取批量ID
                long batchId = message.getId();
                //获取批量的数量
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) {
                    try {
                        Thread.sleep(2000);
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                }else {
                    //如果有数据,处理数据
                    handleEntry(message.getEntries());
                }
                //进行 batch id 的确认。确认之后，小于等于此 batchId 的 Message 都会被确认。
                connector.ack(batchId);
            }
        }
    }

    /**
     * 处理canal server解析binlog获得的实体类信息
     */
    private void handleEntry(List<CanalEntry.Entry> entrys) {
        for (CanalEntry.Entry entry : entrys) {
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                //开启/关闭事务的实体类型，跳过
                continue;
            }
            //RowChange对象，包含了一行数据变化的所有特征
            //比如isDdl 是否是ddl变更操作 sql 具体的ddl sql beforeColumns afterColumns 变更前后的数据字段等等
            CanalEntry.RowChange rowChage;
            try {
                rowChage = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                log.error("parseFrom has an error , data: {}", entry);
                throw new RuntimeException("parseFrom has an error , data:" + entry.toString(), e);
            }
            //获取操作类型：insert/update/delete类型
            CanalEntry.EventType eventType = rowChage.getEventType();
            //打印Header信息
            log.info("binlog[{}:{}] , name[{},{}] , eventType : {}",
                    entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                    entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
                    eventType);
            //判断是否是DDL语句
            if (rowChage.getIsDdl()) {
                log.info("ddl sql: {}", rowChage.getSql());
            }

            //获取RowChange对象里的每一行数据进行处理
            for (CanalEntry.RowData rowData : rowChage.getRowDatasList()) {
                //如果是删除语句
                if (eventType == CanalEntry.EventType.DELETE) {
                    User user = CanalMessageToObject.convertTo(User.class, rowData.getBeforeColumnsList());
                    delete(user);
                } else if (eventType == CanalEntry.EventType.INSERT) {
                    User afterUser = CanalMessageToObject.convertTo(User.class, rowData.getAfterColumnsList());
                    //如果是新增语句
                    insert(afterUser);
                } else {
                    //如果是更新的语句
                    User user = CanalMessageToObject.convertTo(User.class, rowData.getBeforeColumnsList());
                    User afterUser = CanalMessageToObject.convertTo(User.class, rowData.getAfterColumnsList());
                    update(user, afterUser);
                }
            }
        }
    }

    @Resource
    private RedisUtils redisUtils;

    public void insert(User user) {
        log.info("插入新user: {}", user);
        redisUtils.set(RedisConstants.USER_PREFIX + user.getId(), JSON.toJSONString(user),
                RedisConstants.USER_TTL, RedisConstants.TTL_TIME_UNIT);
    }

    public void update(User before, User after) {
        log.info("更新前的user: {}, 更新后的user: {}", before, after);
        redisUtils.delete(RedisConstants.USER_PREFIX + after.getId());
    }

    public void delete(User user) {
        log.info("删除的user: {}", user);
        redisUtils.delete(RedisConstants.USER_PREFIX + user.getId());
    }

}

